/**
 * 
 */
package com.googlecode.xcarve.zk;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.common.config.CarveConstances;
import com.googlecode.xcarve.util.ConfigHolder;

/**
 * @author duxinyun
 * 2014-10-9
 */
public class ZookeeperFactory {
	
	private static final Map<String,ZkClient> registerMap = new ConcurrentHashMap<String, ZkClient>();
	private static Lock lock = new ReentrantLock();
	static final Logger logger = LoggerFactory.getLogger(ZookeeperFactory.class);
	
	public static ZkClient getZKInstance(String domain){
		 if(StringUtils.isBlank(domain)){
			 throw new IllegalArgumentException("param domain cannot be null. please set domain config.");
		 }
		 ZkClient zkClient = registerMap.get(domain);
		 if(zkClient != null){
			 return zkClient;
		 }
		 lock.lock();
		 try {
			zkClient = registerMap.get(domain);
			if( zkClient == null ){
				String address = ConfigHolder.getStringValue(CarveConstances.ZK_ADDRESS);
				zkClient = new ZookeeperClient(address);
				String path = CarveConstances.SEPTOR_PATH+CarveConstances.RPC_NAME+CarveConstances.SEPTOR_PATH+domain; 
				if(!zkClient.exist(path)){
					zkClient.createPath(path);
					logger.info("create application name into zookeeper, this is domain is {}", domain);
				}
				registerMap.put(domain, zkClient);
			}
		} catch (Exception e) {
			throw new IllegalStateException("connect zookeeper fail. so xcarve system cannot be operation.", e);
		}finally{
			lock.unlock();
		}
		return zkClient; 
	}
	
}
